package com.xf.day05
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
/**
 * 分区内聚合： 发生在每个分区内部，使用 createCombiner 和 mergeValue。
 * 分区间聚合： 发生在 shuffle 后的 reduce 阶段，使用 mergeCombiners 合并来自不同分区的累加器
 */
object TestCombineByKey {
  def main(args: Array[String]): Unit = {

    // 创建 SparkConf 对象
    val conf = new SparkConf()
      .setAppName("WordCount")
      .setMaster("local[*]")
      .set("spark.ui.port", "8080")
      .set("spark.driver.host", "127.0.0.1")

    // 创建 SparkContext 对象
    val sc = new SparkContext(conf)
    val rdd11 = sc.parallelize(1 to 9, 3)

    val rdd12 = rdd11.map(item => (item % 3, item))
    val result1 :RDD[(Int, Double)] = rdd12.mapValues(v => v.toDouble)

    // 查看实际分区数
    println(s"实际分区数: ${result1.getNumPartitions}")

    println("=== 分区数据分布 ===")
    result1.mapPartitionsWithIndex { (partitionIndex, iterator) =>
      val partitionData = iterator.toList
      println(s"分区 $partitionIndex: ${partitionData.mkString("[", ", ", "]")}")
      partitionData.iterator
    }.count()

    // 自定义三个函数
    // 1. 创建初始累加器：给定第一个值 v，创建 (sum, count)
    def createCombiner(v: Double): (Double, Int) = {
      (v, 1)
    }

    // 2. 同分区内合并：将新值 v 合并到已有的累加器 c
    def mergeValue(c: (Double, Int), v: Double): (Double, Int) = {
      (c._1 + v, c._2 + 1)
    }

    // 3. 跨分区合并：合并两个累加器 c1 和 c2
    def mergeCombiners(c1: (Double, Int), c2: (Double, Int)): (Double, Int) = {
      (c1._1 + c2._1, c1._2 + c2._2)
    }

    // 使用自定义函数传参
    val result2: RDD[(Int, (Double, Int))] = result1.combineByKey(
      createCombiner,
      mergeValue,
      mergeCombiners
    )

    println( result2.collect().toBuffer )

  }
}
